package com.atguigu.dga230301.governance.assess.calc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.dga230301.ds.bean.TDsTaskDefinition;
import com.atguigu.dga230301.governance.assess.Assessor;
import com.atguigu.dga230301.governance.bean.AssessParam;
import com.atguigu.dga230301.governance.bean.GovernanceAssessDetail;
import com.atguigu.dga230301.governance.util.SqlParser;
import com.atguigu.dga230301.meta.bean.TableMetaInfo;
import com.atguigu.dga230301.meta.constant.MetaConst;
import com.google.common.collect.Sets;
import lombok.Getter;
import org.apache.hadoop.hive.ql.lib.Dispatcher;
import org.apache.hadoop.hive.ql.lib.Node;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.parse.ASTNode;
import org.apache.hadoop.hive.ql.parse.HiveParser;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.text.ParseException;
import java.util.*;

@Component("SIMPLE_PROCESS")
public class SimpleProcessAssess extends Assessor {
    @Override
    protected void assessor(GovernanceAssessDetail governanceAssessDetail, AssessParam assessParam) throws ParseException, org.apache.hadoop.hive.ql.parse.ParseException, SemanticException {
        TableMetaInfo tableMetaInfo = assessParam.getTableMetaInfo();
        String dwLevel = tableMetaInfo.getTableMetaInfoExtra().getDwLevel();
        //如果是ods层的表则不做考评 或者考评日期没运行过的表
        if (dwLevel.equals(MetaConst.DW_LEVEL_ODS)||assessParam.getTDsTaskDefinition()==null) {
            return;
        }
        //1.获取考评表的sql
        TDsTaskDefinition tDsTaskDefinition = assessParam.getTDsTaskDefinition();
        String sql = tDsTaskDefinition.getSql();

        //2.解析sql
        MyDispatcher dispatcher = new MyDispatcher();
        SqlParser.sqlParse(sql, dispatcher);

        //获取库名
        String schemaName = assessParam.getTableMetaInfo().getSchemaName();

        governanceAssessDetail.setAssessComment("涉及的操作："+dispatcher.getProcessSet()+"涉及的过滤字段："+dispatcher.getWhereFieldSet()+"查询的表："+dispatcher.getFromTables());
        //3.将复杂的操作收集到集合中如果集合不为空那么就不是简单加工
        if (dispatcher.processSet.size()>0){
            return;
        }else {
            //4.如果集合为空，获取过滤字段 获取查询的表
            //获取到过滤字段
            Set<String> whereFieldSet = dispatcher.getWhereFieldSet();
            //获取查询的表
            Set<String> fromTables = dispatcher.getFromTables();
            //5.拿着过滤字段和查询的表的分区字段对比看是否过滤字段都是分区字段 如果是的话那么就是简单加工
            //5.1先遍历得到每张表
            for (String fromTable : fromTables) {
                TableMetaInfo fromTableMetaInfo = assessParam.getFullTableNameToMetaInfo().get(schemaName + "." + fromTable);
                //获取到查询表的分区字段
                String partitionColNameJson = fromTableMetaInfo.getPartitionColNameJson();
                //将获取到的json字符串转为list集合
                List<JSONObject> partitionColList = JSON.parseArray(partitionColNameJson, JSONObject.class);

                //遍历过滤字段以及分区字段，然后判断是否与相同
                //创建计数器用来统计过滤字段为分区字段的个数
                Integer sameCount = 0;
                for (String whereField : whereFieldSet) {
                    for (JSONObject partitionJsonObject : partitionColList) {
                        String partititonField = partitionJsonObject.getString("name");
                        if (whereField.equals(partititonField)){
                            sameCount++;
                        }
                    }
                }
                if (sameCount==whereFieldSet.size()){
                    //所有的过滤字段都是分区字段 就是简单加工
                    governanceAssessDetail.setAssessScore(BigDecimal.ZERO);
                    governanceAssessDetail.setAssessProblem("简单加工");
                }
            }
        }

    }

    private static class MyDispatcher implements Dispatcher {
        @Getter
        Set<String> processSet=new HashSet<>();

        @Getter
        Set<String> whereFieldSet=new HashSet<>();

        @Getter
        Set<String> fromTables=new HashSet<>();

        Set<Integer> realProcessSet= Sets.newHashSet(HiveParser.TOK_JOIN,  //join 包含通过where 连接的情况
                HiveParser.TOK_GROUPBY,       //  group by
                HiveParser.TOK_LEFTOUTERJOIN,       //  left join
                HiveParser.TOK_RIGHTOUTERJOIN,     //   right join
                HiveParser.TOK_FULLOUTERJOIN,     // full join
                HiveParser.TOK_FUNCTION,     //count(1)
                HiveParser.TOK_FUNCTIONDI,  //count(distinct xx)
                HiveParser.TOK_FUNCTIONSTAR, // count(*)
                HiveParser.TOK_SELECTDI,  // distinct
                HiveParser.TOK_UNIONALL   // union
        );

        Set<String> operators=Sets.newHashSet("=",">","<",">=","<=" ,"<>" ,"like"); // in / not in 属于函数计算


        /**
         * 首先收集有没有 groupBy join union 这种操作
         *其次 收集所有的过滤字段
         * 最后 收集查询的表
         */
        @Override
        public Object dispatch(Node nd, Stack<Node> stack, Object... nodeOutputs) throws SemanticException {
            ASTNode astNode = (ASTNode) nd;

            //1.判断是否有特殊操作 如果有的话放到processSet这个集合中
            if (realProcessSet.contains(astNode.getType())){
                //有特殊操作
                processSet.add(astNode.getText());
            }

            //2.收集所有的过滤字段
            if (operators.contains(astNode.getText())){
                //证明有做过滤 获取到所有的叶子节点
                ArrayList<Node> children = astNode.getChildren();
                for (Node child : children) {
                    ASTNode astChildNode = (ASTNode) child;
                    //判断是否有 点 意味着字段含带的有表
                    if (astChildNode.getType()== HiveParser.DOT){
                        whereFieldSet.add(astChildNode.getChild(1).getText());
                    }else if (astChildNode.getType()==HiveParser.TOK_TABLE_OR_COL){
                        whereFieldSet.add(astChildNode.getChild(0).getText());
                    }
                }
            }

            //3.收集查询的表
            if (astNode.getType()==HiveParser.TOK_TABREF){
                ASTNode astNodeChild = (ASTNode) astNode.getChild(0); //TOK_TABNAME
                ArrayList<Node> astNodeChildChildren = astNodeChild.getChildren();
                if (astNodeChildChildren.size()==1){
                    //没有库名
                    fromTables.add(astNodeChild.getChild(0).getText());
                }else {
                    //表名带库名  库名在左子叶  表名在右子叶
                    fromTables.add(astNodeChild.getChild(1).getText());
                }
            }


            return null;
        }
    }
}
